package com.hrz.datasync;

import com.aliyun.drc.client.message.DataMessage;
import com.aliyun.drc.clusterclient.ClusterClient;
import com.aliyun.drc.clusterclient.ClusterListener;
import com.aliyun.drc.clusterclient.DefaultClusterClient;
import com.aliyun.drc.clusterclient.RegionContext;
import com.aliyun.drc.clusterclient.message.ClusterMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * @作 者: 侯建军
 * @日 期: 2018/11/04 17:14:02
 * @描 述: 数据同步
 */
public class HrzDataApp {
    private static Logger logger = LoggerFactory.getLogger(HrzDataApp.class);

    // 用户需要替换自己使用的accessKey, accessSecret, subscribeInstanceID
    private static String accessKey = "LTAI5sLGdnQ0RfFe";
    private static String accessSecret = "1FERkTleRcfMq2YmRviZ92Zpbxdx1Q";
    private static String subscribeInstanceID = "dtsto1unftk0j08";

    /**
     * 功能描述:
     * 命令行输入参数,通过命令行传入key与id
     *
     * @参 数:
     * @返 回:
     * @作 者: 侯建军
     * @日 期: 2018/11/04 05:43:40
     */
//    private static void parsePromptArgs(String args[]) {
//        for (int i = 0; i < args.length; i++) {
//            String prompt = args[i++];
//            if (i >= args.length) {
//                break;
//            } else if (prompt.equals("--accessKey")) {
//                accessKey = args[i];
//            } else if (prompt.equals("--accessSecret")) {
//                accessSecret = args[i];
//            } else if (prompt.equals("--subscribeInstanceID")) {
//                subscribeInstanceID = args[i];
//            } else {
//                System.err.println("Usage: java -jar demo.jar --accessKey accessKey --accessSecret accessSecret --subscribeInstanceID subscribeInstanceID");
//                System.exit(1);
//            }
//        }
//    }
    public static void main(String[] args) throws Exception {

        logger.info("start");
        //创建上下文
        RegionContext context = new RegionContext();
        //连接DTS 公网为true
        context.setUsePublicIp(false);
        //阿里云key
        context.setAccessKey(accessKey);
        //
        context.setSecret(accessSecret);
        // 创建一个消费客户端
        final ClusterClient client = new DefaultClusterClient(context);

        // 创建订阅监听
        ClusterListener listener = new ClusterListener() {
            @Override
            public void notify(List<ClusterMessage> messages) throws Exception {
                System.out.println("消息数："+messages.size());
                for (ClusterMessage message : messages) {
                    //打印订阅到的数据
//                    System.out.println(message.getRecord() + ":" + message.getRecord().getTablename() + ":"
//                            + message.getRecord().getOpt());
//                    String siteid = message.getRecord().getAttribute("siteid");
//                    System.out.println(siteid);
                    DataMessage.Record record = message.getRecord();
//                    //数据读取
//                    DataMessage.Record.Type type=record.getOpt();
//                    String ddl_string;
//                    if(type.equals(DataMessage.Record.Type.DDL)){
//                        List<DataMessage.Record.Field> fields = record.getFieldList();
//                        ddl_string = fields.get(0).getValue().toString();
//                        System.out.println(ddl_string);
//                    }
                    System.out.println("数据获取开始");
                    StringBuilder insert_string = new StringBuilder();
                    DataMessage.Record.Type type = record.getOpt();
                    DataMessage.Record.Field field;
                    StringBuilder FieldName = new StringBuilder();
                    StringBuilder FieldValue = new StringBuilder();
                    System.out.println("消息类型："+type);
                    //如果是insert语句
                    if (type.equals(DataMessage.Record.Type.INSERT)) {
                        int i = 0;
                        List<DataMessage.Record.Field> fields = record.getFieldList();
                        for (; i < fields.size(); i++) {
                            field = fields.get(i);
                            FieldName.append('`' + field.getFieldname().toLowerCase() + '`');
                            FieldValue.append("'" + field.getValue() + "'");
                            if (i != fields.size() - 1) {
                                FieldName.append(',');
                                FieldValue.append(',');
                            }
                        }
                        insert_string.append("insert " + record.getTablename() + "(" + FieldName.toString() + ") values(" + FieldValue.toString() + ");");
                       logger.info(insert_string.toString());
                        //消费完数据后向DTS汇报ACK
                        message.ackAsConsumed();
                    }
                    System.out.println("数据获取结束");
                }
            }
            @Override
            public void noException(Exception e) {
                e.printStackTrace();
            }
        };
        //添加监听
        client.addConcurrentListener(listener);
        //设置订阅请求通道ID，为需要订阅的数据订阅实例 ID，需要到 数据传输 DTS 控制台 中获取
        client.askForGUID(subscribeInstanceID);
        try {
            //启动应用
            client.start();
        } catch (Exception ex) {

            client.start();
        }
        System.out.println("连接成功！");
    }
}
